package com.o2o.cleaning.month.platform.ebusiness_plat.brand_modular.utils

import org.apache.spark.sql.{DataFrame, SparkSession}

class brand_state_util {

  def brand_state(brand_stap_2:DataFrame, spark:SparkSession, platformName:String,year : Int,month: Int): DataFrame ={
//    新增品牌
    brand_stap_2.registerTempTable("addNewBrand")


//   将天猫，京东，苏宁的品牌表作为主表，新平台的品牌表向其靠拢
    val obs = s"s3a://o2o-dimension-table/brandName_table/${year}/${month}/"
    val state_1 = spark.read.json(obs+"tmall",obs+"jd",obs+"suning")
    state_1.registerTempTable("brand_stateTables")

//    新增品牌数据打上品牌来源国标签
    val stateIsJoin = spark.sqlContext.sql(
      """
        |select a.*,
        |b.brand_state as state,
        |b.brand_isLaoZiHao laozihao
        |from addNewBrand a
        |join
        |(select firstCategoryId,brandName_cn,brand_state from brand_stateTables group by firstCategoryId,brandName_cn,brand_state)b
        |on a.brandName_cn = b.brandName_cn and a.firstCategoryId = b.firstCategoryId
      """.stripMargin)
      .drop("brand_state","brand_isLaoZiHao")
      .withColumnRenamed("state", "brand_state")
      .withColumnRenamed("laozihao", "brand_isLaoZiHao")


    val stateNotJoin = spark.sqlContext.sql(
      """
        |select a.*,
        |'0' state,
        |'0' laozihao
        |from addNewBrand a
        |left join
        |(select firstCategoryId,brandName_cn,brand_state from brand_stateTables group by firstCategoryId,brandName_cn,brand_state)b
        |on a.brandName_cn = b.brandName_cn and a.firstCategoryId = b.firstCategoryId
        |where b.firstCategoryId is null and b.brandName_cn is null
      """.stripMargin)
      .drop("brand_state","brand_isLaoZiHao")
      .withColumnRenamed("state", "brand_state")
      .withColumnRenamed("laozihao", "brand_isLaoZiHao")


    val stateAll = stateIsJoin.toJSON.rdd.union(stateNotJoin.toJSON.rdd)

    val stateResult = spark.read.json(stateAll)
      .dropDuplicates("brandCcId")

    return stateResult
  }
}
